package com.guojizhuang.join;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class JoinReducer extends Reducer<LongWritable, Text, LongWritable,Text> {

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values,Context context) throws IOException, InterruptedException {


        // 由于values中包含了student和attendance数据，所以执行第一次循环：
        // 把student和attendance分别放到不同的list中
        StringBuilder stringBuilder = new StringBuilder();
        boolean flag = false;

        for (Text val : values) {
            // 检查当前Text值是否以"location"开头
            if (val.toString().startsWith("location")) {
                flag = true;
            }

            // 提取以":"分隔的第二部分，并添加到StringBuilder中
            String[] parts = val.toString().split(":");
            if (parts.length > 1) {
                stringBuilder.append(parts[1]);
            }

            // 添加逗号，但不加在最后一个元素后面
            if (!stringBuilder.toString().endsWith(",")) {
                stringBuilder.append(",");
            }
        }
        if (flag){
            // 去掉最后的逗号，并返回结果字符串
            String concatenatedString = stringBuilder.toString().trim().replaceAll(",$", "");
            context.write(key, new Text(concatenatedString));
        }

    }
}
